package com.gd.data.platform.flink.utils;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.HashMap;
import java.util.Map;

import com.fasterxml.jackson.databind.ObjectMapper;

import lombok.extern.slf4j.Slf4j;

/**
 * Flink工具类
 * 提供Flink作业中常用的工具方法
 * 
 * @author gd-data-platform
 * @since 1.0.0
 */
@Slf4j
public class FlinkUtils {

    private static final ObjectMapper objectMapper = new ObjectMapper();
    private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    /**
     * 解析JSON字符串为Map
     */
    @SuppressWarnings("unchecked")
    public static Map<String, Object> parseJson(String json) {
        try {
            return objectMapper.readValue(json, Map.class);
        } catch (Exception e) {
            log.error("解析JSON失败：{}", json, e);
            return new HashMap<>();
        }
    }

    /**
     * 将对象转换为JSON字符串
     */
    public static String toJson(Object obj) {
        try {
            return objectMapper.writeValueAsString(obj);
        } catch (Exception e) {
            log.error("转换为JSON失败", e);
            return "{}";
        }
    }

    /**
     * 获取当前时间字符串
     */
    public static String getCurrentTimeString() {
        return LocalDateTime.now().format(formatter);
    }

    /**
     * 创建实时指标数据
     */
    public static Map<String, Object> createRealtimeMetric(String metricType, String metricName,
            Double metricValue, String dimensions) {
        Map<String, Object> metric = new HashMap<>();
        metric.put("metricType", metricType);
        metric.put("metricName", metricName);
        metric.put("metricValue", metricValue);
        metric.put("windowStartTime", getCurrentTimeString());
        metric.put("windowEndTime", getCurrentTimeString());
        metric.put("dimensions", dimensions);
        metric.put("createTime", getCurrentTimeString());
        return metric;
    }

    /**
     * 检查告警条件
     */
    public static boolean checkAlertCondition(String metricType, Double metricValue, Double threshold) {
        if (metricValue == null || threshold == null) {
            return false;
        }

        switch (metricType) {
            case "call_count":
                return metricValue > threshold;
            case "failure_rate":
                return metricValue > threshold;
            case "download_count":
                return metricValue > threshold;
            default:
                return false;
        }
    }

    /**
     * 格式化时间窗口
     */
    public static String formatTimeWindow(LocalDateTime startTime, LocalDateTime endTime) {
        return startTime.format(formatter) + " 到 " + endTime.format(formatter);
    }
}
